package cn.jly.flink.window;

import cn.jly.flink.entity.SensorReading;
import cn.jly.flink.source2sink.datastream.source.CustomSource;
import cn.jly.flink.utils.FlinkUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

/**
 * @PackageName cn.jly.flink.window
 * @ClassName CountWindowDemo
 * @Description 计数窗口测试
 * @Author 姬岚洋
 * @Date 2021/2/1 上午 10:17
 */
public class CountWindowDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = FlinkUtils.getStreamExecutionEnv();

        env.addSource(new CustomSource.CustomSensorReadingSourceFunction("sensorReading数据源"))
                .filter(new FilterFunction<SensorReading>() {
                    @Override
                    public boolean filter(SensorReading value) throws Exception {
                        return "sensor_4".equals(value.getId());
                    }
                })
                .keyBy(new KeySelector<SensorReading, String>() {
                    @Override
                    public String getKey(SensorReading value) throws Exception {
                        return value.getId();
                    }
                })
                // 每来10个触发一次process处理
                .countWindow(10)
                .process(new ProcessWindowFunction<SensorReading, Tuple2<String, Integer>, String, GlobalWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<SensorReading> elements, Collector<Tuple2<String, Integer>> out) throws Exception {
                        out.collect(Tuple2.of(s, CollectionUtils.size(elements)));
                    }
                })
                .print();

        FlinkUtils.executeStream(env, "CountWindowDemo");
    }
}
